#include "include/sp.h"

#include <sys/types.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#define int32u unsigned int


#define MAX_MESSLEN     102400
#define MAX_VSSETS      10
#define MAX_MEMBERS     100

#define SEND_BURST_SIZE 3
#define RECV_BURST_SIZE (numMachines*5)
#define MAX_NUM_SEND 500
#define TYPE_MCAST_MSG 1
#define TYPE_FINISH_MSG 2

#define HEADER_SIZE (sizeof(Header))
#define PAYLOAD_SIZE 1000
#define PACKET_SIZE (PAYLOAD_SIZE + HEADER_SIZE)
#define NAME_LENGTH 80

#define TRUE 1
#define FALSE 0

static char User[80];
static char Spread_name[80];

static char Private_group[MAX_GROUP_NAME];
static mailbox Mbox;

static int To_exit = 0;
static char Group_name[80];

int totalNumPktsToSend, machineIdx, numMachines;
int curProcessIdx = 0;
int curSentInWindow = 0;
unsigned int numProcessesAlive;
int isFinishMsgSent = FALSE;
int numRecvd = 0;
FILE *fd;
sp_time startTime;



/* Data Structures */
struct Header{      
        int pid;    
        int idx;
        int randomData;
};

typedef struct Header Header;

static void Read_messages();
static void Send_messages();
static void Usage(int argc, char *argv[]);
static void Bye();

void myprintf (char *str, ...) {

}

/* Generate random number */
unsigned long bigrandom() {     
    int i;
    unsigned long n = 0;
    /* First digit must be non-zero */
    do {
        n = rand() % 10;
    } while (n == 0);
    for (i = 1; i < 6; i++) {
        n *= 10;
        n += rand() % 10;
    }
    return n;
}


int main(int argc, char *argv[]) {
    int ret;
    int mver, miver, pver;
    sp_time test_timeout;
    static char mess[MAX_MESSLEN];
    char sender[MAX_GROUP_NAME];
    char target_groups[MAX_MEMBERS][MAX_GROUP_NAME];
    membership_info memb_info;
    int num_groups, i;
    int service_type;
    int16 mess_type;
    int endian_mismatch;

    char filePath[NAME_LENGTH];


    test_timeout.sec = 5;
    test_timeout.usec = 0;

    Usage(argc, argv);
    totalNumPktsToSend = atoi(argv[1]);
    machineIdx = atoi(argv[2]);
    numMachines = atoi(argv[3]);
    numProcessesAlive = numMachines;
    strncpy(filePath, argv[2], 1); /* ProcessIdx is only 1 byte long(0 to 9) */
    strncat(filePath, ".out", sizeof(".out"));
    if((fd = fopen(filePath, "w")) == NULL){
        perror("fopen");
        exit(1);
    }

    if (!SP_version(&mver, &miver, &pver)) {
        printf("main: Illegal variables passed to SP_version()\n");
        Bye();
    }
    printf("Spread library version is %d.%d.%d\n", mver, miver, pver);

    strcpy(User, "para");   /* Set User as para */
    strcpy(Group_name, "paragrp");  /* Set groupname as paragrp */
    ret = SP_connect_timeout(Spread_name, User, 0, 1, &Mbox, Private_group, test_timeout);
    if (ret != ACCEPT_SESSION) {
        SP_error(ret);
        Bye();
    }
    printf("User: connected to %s with private group %s\n", Spread_name, Private_group);

    E_init();

    ret = SP_join(Mbox, Group_name);
    if (ret < 0) SP_error(ret);


    for (;numMachines != num_groups;) {
        service_type = 0;
        ret = SP_receive(Mbox, &service_type, sender, 100, &num_groups, target_groups,
                &mess_type, &endian_mismatch, sizeof (mess), mess);
        printf("\n============================\n");
        if (ret < 0) {
            if ((ret == GROUPS_TOO_SHORT) || (ret == BUFFER_TOO_SHORT)) {
                service_type = DROP_RECV;
                printf("\n========Buffers or Groups too Short=======\n");
                ret = SP_receive(Mbox, &service_type, sender, MAX_MEMBERS, &num_groups, target_groups,
                        &mess_type, &endian_mismatch, sizeof (mess), mess);
            }
        }

        if (ret < 0) {
            if (!To_exit) {
                SP_error(ret);
                printf("\n============================\n");
                printf("\nBye.\n");
            }
            exit(0);
        }

        if (Is_membership_mess(service_type)) {
            ret = SP_get_memb_info(mess, service_type, &memb_info);
            if (ret < 0) {
                printf("BUG: membership message does not have valid body\n");
                SP_error(ret);
                exit(1);
            }
            if (Is_reg_memb_mess(service_type)) {
                printf("Received REGULAR membership for group %s with %d members, where I am member %d:\n",
                        sender, num_groups, mess_type);
                for (i = 0; i < num_groups; i++)
                    printf("\t%s\n", &target_groups[i][0]);
                printf("grp id is %d %d %d\n", memb_info.gid.id[0], memb_info.gid.id[1], memb_info.gid.id[2]);

                if (Is_caused_join_mess(service_type)) {
                    printf("Due to the JOIN of %s\n", memb_info.changed_member);
                }
            }
        }
    }

    printf("Start sending now..\n");

    startTime = E_get_time();
    for(;;) {
        Send_messages();
        Read_messages();
    }

    return (0);
}

/* Send the messages */
static void Send_messages() {
    int i, ret;
    char sendPkt[PACKET_SIZE];
    Header *sendHdr = (Header*)sendPkt;
    myprintf ("In send\n");
    while(curSentInWindow < MAX_NUM_SEND && curProcessIdx < totalNumPktsToSend) {
        /* Send Burst Size number of Packets */
        for(i = 0; i < SEND_BURST_SIZE && curSentInWindow < MAX_NUM_SEND && curProcessIdx < totalNumPktsToSend; i++, curSentInWindow++, curProcessIdx++) {
                /* Construct Packet */
                sendHdr->pid = machineIdx;
                sendHdr->idx = curProcessIdx;
                sendHdr->randomData = bigrandom();

                ret = SP_multicast(Mbox, AGREED_MESS, Group_name, TYPE_MCAST_MSG, PACKET_SIZE, sendPkt);
                if (ret < 0) {
                    SP_error(ret);
                    Bye();
                }
                myprintf ("--> Sent message %d (curSentInWindow %d) (total %d)\n", i + 1, curSentInWindow, curProcessIdx);
        }

        /* Check if there are any messages recieved */
        ret = SP_poll(Mbox);
        myprintf ("Polling sais: %d, (curSentInWindow %d) \n", ret, curSentInWindow);
        if(ret) {
            break;
        }
    }
    myprintf ("Out from While\n");

    if(curProcessIdx == totalNumPktsToSend && !isFinishMsgSent) {
        printf ("Sending finish message\n");
        ret = SP_multicast(Mbox, AGREED_MESS, Group_name, TYPE_FINISH_MSG, PACKET_SIZE, sendPkt);
        isFinishMsgSent = TRUE;
    }

}

/* Receive the messages */
static void Read_messages() {

    char recvPkt[PACKET_SIZE];
    Header *recvHdr = (Header*)recvPkt;

    char sender[MAX_GROUP_NAME];
    char target_groups[MAX_MEMBERS][MAX_GROUP_NAME];
    int num_groups;
    int service_type;
    int16 mess_type;
    int endian_mismatch;

    int isAnyMessagePresent, ret;

    isAnyMessagePresent = SP_poll(Mbox);
    myprintf ("In Recv (%d)\n", isAnyMessagePresent);

    while(isAnyMessagePresent) { 
        service_type = 0;

        ret = SP_receive(Mbox, &service_type, sender, 100, &num_groups, target_groups,
                &mess_type, &endian_mismatch, sizeof (recvPkt), recvPkt);
        myprintf ("<-- Recieved message (curSentInWindow %d) (numRecvd %d) \n",curSentInWindow, numRecvd+1);

        if (ret < 0) {
            if ((ret == GROUPS_TOO_SHORT) || (ret == BUFFER_TOO_SHORT)) {
                service_type = DROP_RECV;
                printf("\n========Buffers or Groups too Short=======\n");
                ret = SP_receive(Mbox, &service_type, sender, MAX_MEMBERS, &num_groups, target_groups,
                        &mess_type, &endian_mismatch, sizeof (recvPkt), recvPkt);
            }
        }
        if (ret < 0) {
            if (!To_exit) {
                SP_error(ret);
                printf("\n============================\n");
                printf("\nBye.\n");
            }
            exit(0);
        }

        numRecvd++;

        if (Is_regular_mess(service_type)) {
            if(mess_type == TYPE_MCAST_MSG) { /* Check if it is a multicast message */
                fprintf(fd, "%5d, %6d, %6d\n", recvHdr->pid, recvHdr->idx, recvHdr->randomData);    /* Write to the file */
            } else if(mess_type == TYPE_FINISH_MSG) { /* Check if it is a FINISH message */
                numProcessesAlive--;
            }

        } else if (Is_membership_mess(service_type)) {
            /* Ignore this */
        }

        isAnyMessagePresent = SP_poll(Mbox);
        myprintf ("In recv:while (%d)\n", isAnyMessagePresent);

    }

    /* Check if the protocol has reached termination */
    if(numProcessesAlive == 0) {
        printf("Quitting\n");
        fclose(fd);
        Bye();
    }

    /* Adjust the window counter */
    curSentInWindow -= (int)(numRecvd/numProcessesAlive);
    myprintf ("Recalculated (curSentInWindow %d) (temp %d) (numRecvd %d) \n",curSentInWindow, numRecvd/numProcessesAlive, numRecvd);
    
    numRecvd = numRecvd % numProcessesAlive;

    if(curSentInWindow < 0) {
        curSentInWindow = 0;
    }
    myprintf ("Before returning from recieved messages (curSentInWindow %d) (numRecvd %d) \n",curSentInWindow, numRecvd);

}

static void Usage(int argc, char *argv[]) {
    sprintf(User, "user");
    sprintf(Spread_name, "4803@localhost");
    if (argc != 4) {
        printf("Usage: mcast <num_of_messages> <process_index> <num_of_processes> \n");
        exit(0);
    }
}

static void Bye() {
    To_exit = 1;
    printf("\nBye.\n");
    SP_disconnect(Mbox);
    printf("Total time taken = %ld ms\n", (E_sub_time(E_get_time(), startTime).sec*1000) + (E_sub_time(E_get_time(), startTime).usec/1000));
    exit(0);
}
